package com.chukun.flink.stream.window.time;

import com.chukun.flink.stream.util.TimeUtils;
import com.chukun.flink.stream.window.process.bean.EventBean;
import com.chukun.flink.stream.window.source.SourceWithTimestampsWatermarkers;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.Calendar;
import java.util.Date;

/**
 * @author chukun
 * @version 1.0.0
 * @description 基于事件时间的操作
 * @createTime 2022年05月24日 22:46:00
 */
public class EventTimeOperator {

    public static void main(String[] args) throws Exception {

        // 当前时间在 30~35秒之间，以及0时启动处理程序，以保证生成的结果不受窗口对齐机制的影响
        while (true) {
            Calendar instance = Calendar.getInstance();
            instance.setTime(new Date());
            int offset = instance.get(Calendar.SECOND);
            if (offset <=0) {
                break;
            } else if (offset >= 30 && offset<=35) {
                break;
            }
        }

        System.out.println("start time : "+ TimeUtils.getHHmmss(new Date()));

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<EventBean> streamSource = env.addSource(new SourceWithTimestampsWatermarkers(10000));


        SingleOutputStreamOperator<EventBean> reduce = streamSource
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(30)))
                .reduce(new ReduceFunction<EventBean>() {
                    @Override
                    public EventBean reduce(EventBean eventBean, EventBean t1) throws Exception {
                        eventBean.getList().add(t1.getList().get(0));
                        return eventBean;
                    }
                });

        // 设置watermarker的刷新时间间隔
        env.getConfig().setAutoWatermarkInterval(5000);

        // 加载没有发送watermarker的数据源
        DataStreamSource<EventBean> streamWithoutTimestampSource = env.addSource(new SourceWithTimestampsWatermarkers(10000));

        // 原始数据处理
        SingleOutputStreamOperator<EventBean> handleEventTimeStream = streamWithoutTimestampSource
                .assignTimestampsAndWatermarks(
                        // 生成水印 watermarker
                        WatermarkStrategy.forGenerator((ctx) -> new EventWatermarkerGenerator())
                                // 抽取事件产生的时间
                                .withTimestampAssigner((ctx) -> new EventTimestampExtractor()));

        // 根据自定义生成watermarker与事件时间的策略聚合数据
        SingleOutputStreamOperator<EventBean> copyedReduce = handleEventTimeStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(30)))
                .reduce(new ReduceFunction<EventBean>() {
                    @Override
                    public EventBean reduce(EventBean eventBean, EventBean t1) throws Exception {
                        eventBean.getList().add(t1.getList().get(0));
                        return eventBean;
                    }
                });


        reduce.print("with-source-eventTime");

        copyedReduce.print("without-source-eventTime");

        env.execute("EventTimeOperator");
    }
}
